package net.bwie.realtime.warehouse.DWD;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.DWD
 * @Author: zhangleqing
 * @CreateTime: 2025-09-01  20:09
 * @Description: TODO dwd_refund_reason(退款原因汇总表)
 * 相关表：refund_detail（退款明细信息表）  order_base（订单基础信息表）  consumer_behavior（消费者行为表）
 * @Version: 1.0
 */
public class DwdRefundReason {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 数据清洗+加工
        Table resultTable = handle(tableEnv);

//        resultTable.execute().print();

        // 4 映射表创建
        createView(tableEnv);

        // 5 输出数据
        saveToSink(tableEnv, resultTable);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("DwdRefundReason", resultTable);
        tableEnv.executeSql(
                "insert into dwd_refund_reason " +
                        "select * from DwdRefundReason"
        ).print();
    }

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_refund_reason (\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    order_amount DOUBLE,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    \n" +
                        "    refund_id STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    \n" +
                        "    behavior_id STRING,\n" +
                        "    behavior_type STRING,\n" +
                        "    PRIMARY KEY (order_id) NOT ENFORCED\n" +
                        ") WITH (\n" +
//                        "     'connector' = 'kafka',\n" +
//                        "     'topic' = 'dwd_refund_reason',\n" +
//                        "     'properties.bootstrap.servers' = 'node101:9092',\n" +
//                        "     'format' = 'json'\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dwd_retreat_behavior',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );
    }


/*
SELECT
    o.order_id,                -- 订单ID
    o.user_id,                 -- 消费者ID
    o.product_id,              -- 商品ID
    o.order_amount,            -- 订单总金额
    o.pay_time,                -- 订单支付时间
    r.refund_id,               -- 退款单号
    r.refund_amount,           -- 实际退款金额
    cb.behavior_id,            -- 消费者行为ID
    cb.behavior_type           -- 行为类型
FROM ods_order_base o          -- 订单主表
-- 左连接退款表，通过订单ID关联
LEFT JOIN ods_refund_detail r
    ON o.order_id = r.order_id
    AND r.apply_time >= o.pay_time  -- 确保退款在支付之后
    AND r.product_id = o.product_id -- 确保商品匹配
-- 左连接消费者行为表，通过用户ID关联
LEFT JOIN ods_consumer_behavior cb
    ON o.user_id = cb.user_id
-- 过滤掉支付时间为空的无效订单
WHERE o.pay_time IS NOT NULL and o.order_id is not null
and r.refund_id is not null and o.product_id is not null
 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "SELECT\n" +
                        "    o.order_id,                -- 订单ID\n" +
                        "    o.user_id,                 -- 消费者ID\n" +
                        "    o.product_id,              -- 商品ID\n" +
                        "    o.order_amount,            -- 订单总金额\n" +
                        "    o.pay_time,                -- 订单支付时间\n" +
                        "    r.refund_id,               -- 退款单号\n" +
                        "    r.refund_amount,           -- 实际退款金额\n" +
                        "    cb.behavior_id,            -- 消费者行为ID\n" +
                        "    cb.behavior_type           -- 行为类型\n" +
                        "FROM ods_order_base o          -- 订单主表\n" +
                        "-- 左连接退款表，通过订单ID关联\n" +
                        "LEFT JOIN ods_refund_detail r\n" +
                        "    ON o.order_id = r.order_id\n" +
                        "    AND r.apply_time >= o.pay_time  -- 确保退款在支付之后\n" +
                        "-- 左连接消费者行为表，通过用户ID关联\n" +
                        "LEFT JOIN ods_consumer_behavior cb\n" +
                        "    ON o.user_id = cb.user_id\n" +
                        "-- 过滤掉支付时间为空的无效订单\n" +
                        "WHERE o.pay_time IS NOT NULL and o.order_id is not null\n" +
                        "and r.refund_id is not null"
        );
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        // 1 refund_detail（退款明细信息表）
        tableEnv.executeSql(
                "create table ods_refund_detail (\n" +
                        "    refund_detail_id STRING,\n" +
                        "    refund_id STRING,\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    apply_time TIMESTAMP(3),\n" +
                        "    refund_status STRING,\n" +
                        "    finish_time TIMESTAMP(3),\n" +
                        "    product_id STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    user_fill_reason STRING,\n" +
                        "    refund_scene STRING,\n" +
                        "    WATERMARK FOR apply_time AS apply_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-refund-detail-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-refund-detail-log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 2 order_base（订单基础信息表）
        tableEnv.executeSql(
                "create table ods_order_base (\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    shop_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "    click_ship_time TIMESTAMP(3),\n" +
                        "    order_amount DECIMAL(10,2),\n" +
                        "    order_status STRING,\n" +
                        "    WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-order-base-Log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-order-base-Log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 3 consumer_behavior（消费者行为表）
        tableEnv.executeSql(
                "create table ods_consumer_behavior (\n" +
                        "    behavior_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    behavior_type STRING,\n" +
                        "    behavior_time TIMESTAMP(3),\n" +
                        "    behavior_scenario STRING,\n" +
                        "    WATERMARK FOR behavior_time AS behavior_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-consumer-behavior-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-consumer-behavior-log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
    }


    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
